package com.alan.datasync.converter;

import com.alan.datasync.enums.MessageActionEnum;
import com.alan.datasync.wrapper.MessageEventWrapper;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.FastDateFormat;

import java.text.ParseException;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author: li_jl
 * @Description: json转换器
 * @Date: Created in 2020/6/8 4:55 下午
 */
public class OggJsonConverter {

    private OggJsonConverter() {
    }

    public static final FastDateFormat OPERATION_TIME_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd hh:mm:ss.SSS");
    public static final FastDateFormat CURRENT_TIME_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd'T'hh:mm:ss.SSS");

    public static final int ST_TMP_LENGTH = 2;

    private static final Map<String, MessageActionEnum> EVENT_MAPPING = new HashMap<String, MessageActionEnum>() {
    };

    static {
        EVENT_MAPPING.put("I", MessageActionEnum.INSERT);
        EVENT_MAPPING.put("U", MessageActionEnum.UPDATE);
        EVENT_MAPPING.put("D", MessageActionEnum.DELETE);
    }

    public static MessageEventWrapper convert(String jsonMessage) throws ParseException {

        JSONObject obj = JSON.parseObject(jsonMessage);
        MessageActionEnum action = EVENT_MAPPING.getOrDefault(obj.getString("op_type"), MessageActionEnum.UNKNOWN);
        // 不能解析的事件跳过
        if (null == action || action == MessageActionEnum.UNKNOWN) {
            return null;
        }
        MessageEventWrapper event = new MessageEventWrapper();

        // 库表
        String schemaAndTable = obj.getString("table");
        String[] stTmp = null != schemaAndTable ? schemaAndTable.split("\\.") : null;
        if (null != stTmp && stTmp.length == ST_TMP_LENGTH) {
            event.setSchema(stTmp[0]);
            event.setTable(stTmp[1]);
        }
        // 操作类型
        event.setOpType(action);

        String opTs = obj.containsKey("op_ts") ? obj.getString("op_ts") : null;
        if (StringUtils.isNotEmpty(opTs)) {
            event.setOperationTime(OPERATION_TIME_FORMAT.parse(opTs.substring(0, opTs.length() - (opTs.split("\\.")[1].length() - 3))));
        }

        String currentTs = obj.containsKey("current_ts") ? obj.getString("current_ts") : null;
        if (StringUtils.isNotEmpty(currentTs)) {
            event.setCurrentTime(CURRENT_TIME_FORMAT.parse(currentTs.substring(0, currentTs.length() - 3)));
        }

        JSONArray primaryKeys = obj.containsKey("primary_keys") ? obj.getJSONArray("primary_keys") : null;
        if (null != primaryKeys) {
            event.setPrimaryKeys(primaryKeys.toJavaList(String.class));
        }

        event.setBefore(obj.getObject("before", Map.class));
        event.setAfter(obj.getObject("after", Map.class));
        Calendar cal = Calendar.getInstance();
        event.setConsumedTime(cal.getTime());
        event.setFullColumnValue(false);

        return event;
    }
}
